package com.atguigu.chapter10;

import com.atguigu.Bean.OrderEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @ClassName: Flink02_OrderWatch_CEP_project
 * @Description:
 * @Author: kele
 * @Date: 2021/4/12 9:20
 *
 *
 * 基于CEP的订单实时监控
 *
 * 1、获取数据，封装对象，添加水印，按照订单id进行keyby
 * 2、使用cep进行匹配，
 *
 *
 **/
public class Flink02_OrderWatch_CEP_project {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment exec
                = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        exec.setParallelism(2);

        KeyedStream<OrderEvent, Long> ds = exec.readTextFile("in/OrderLog.csv")
                .map((line -> {
                    String[] datas = line.split(",");
                    return new OrderEvent(
                            Long.valueOf(datas[0]),
                            datas[1],
                            datas[2],
                            Long.parseLong(datas[3]) * 1000);
                }))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {

                                    @Override
                                    public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                                        return element.getEventTime();
                                    }
                                })
                )

                .keyBy(OrderEvent::getOrderId);

/**
 * 使用CEP进行匹配
 *  匹配第一次是create，第二次是pay，之间时间间隔是15min，
 *
 *  将不符合条件的输出到侧输出流中
 *  CEP.pattern(ds, order)
 *                 .select(
 *                      侧输出流的名字
 *                      侧输出流的类型
 *                      主流的输出
 *                 )
 *
 */
        Pattern<OrderEvent, OrderEvent> order = Pattern
                .<OrderEvent>begin("create")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return "create".equalsIgnoreCase(value.getEventType());
                    }
                })
                .next("pay")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return "pay".equalsIgnoreCase(value.getEventType());
                    }
                })
                .within(Time.minutes(15));

        SingleOutputStreamOperator<String> result = CEP.pattern(ds, order)
                .select(
                        new OutputTag<String>("later") {},
                        new PatternTimeoutFunction<OrderEvent, String>() {
                            @Override
                            public String timeout(Map<String, List<OrderEvent>> map, long l) throws Exception {
                                 return map/*.get("create").toString() + map.get("pay")*/.toString();
//                                return map.get("pay").toString();  //pay可能匹配不到，出现空指针异常
                            }
                        },
                        new PatternSelectFunction<OrderEvent, String>() {
                            @Override
                            public String select(Map<String, List<OrderEvent>> map) throws Exception {
                                return map.toString();
                            }
                        }
                );

        result.print("normal");
        result.getSideOutput(new OutputTag<String>("later") {}).print("later");


        exec.execute();


    }


}
